package com.ch3cooh.registry.service;

import com.alibaba.fastjson.JSON;
import com.ch3cooh.registry.pojo.RegisterInfo;
import com.ch3cooh.registry.util.StringUtils;
import com.ch3cooh.rpc.component.LocalServerManager;
import com.ch3cooh.rpc.util.SpringUtils;
import lombok.extern.java.Log;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
@Log
public class ZkService {

    @Autowired
    CuratorFramework client;
    public CuratorFramework  getClient(){
        return client;
    }

    @Autowired
    LocalServerManager localServerManager;

    /**
     * 设置节点和节点数据
     * @param key
     * @param data
     * @return
     */
    public  String set(String key,byte[] data){
        try {
            key = formatePath(key);
            if( checkExistsNode(key) ){
                return client.create().forPath(key,data);
            }
            client.setData().forPath( key , data);
            return "ok";
        } catch (Exception e) {
            e.printStackTrace();
            return "nil";
        }
    }

    /**
     * 创建一个临时节点
     * @param node  节点名
     * @param data   节点数据
     * @return
     */
    public boolean createEphemeralNode(String node, String data){
        String result = null;
        try {
            result = client.create().creatingParentContainersIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(node,data.getBytes() );
        } catch (Exception e) {
            e.printStackTrace();
        }
        return result!=null;
    }

    /**
     * 判断一个节点是否存在
     * @param path  节点名
     * @return  true表示存在；false表示不存在
     */
    public boolean checkExistsNode(String path){
        try {
            boolean res = !(client.checkExists().forPath(path) == null);
            return res;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 获取节点对应的节点数据
     * @param key
     * @return
     */
    public  String get(String key ){
        try {
            return new String ( client.getData().forPath(formatePath(key) ),"utf-8");
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 查找一个节点下的所有节点(只查找一层)
     * @param node
     * @return
     */
    public Map<String, List<String>> listOneLayer(String node){
        try {
            List<String> childNodes = client.getChildren().forPath(formatePath(node));
            if(childNodes.size() >0){
                Map<String,List<String>> collect = new HashMap<>(childNodes.size());
              //  Map<String, String> collect = childs.stream().collect(Collectors.toMap(e -> e, e -> StringUtils.newString(get(e),"utf-8")));
                for (String e : childNodes) {
                    String childPaht = node+formatePath(e);
                    List<String> childchildNodes = client.getChildren().forPath(childPaht);
                    List<String> childchildData = new ArrayList<>(childchildNodes.size());
                    for(String e2 : childchildNodes){
                        String data = get( node+formatePath(e) +formatePath(e2));
                        if(data != null){
                            childchildData.add(data);
                        }
                    }
                    collect.put(e,childchildData);
                }
                return collect;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new HashMap<>();
    }

    /**
     * 添加zk监听器，订阅指定节点以及其下所有子节点的变化
     * @param nodeName  要监听的节点名字
     * @return
     */
    public  boolean addListener(String nodeName){
        TreeCache treeCache = new TreeCache(client,formatePath(nodeName));
        treeCache.getListenable().addListener( (c,e)->{
            ChildData data = e.getData();
            String dataStr;
            if(data!=null && !StringUtils.isBlack(dataStr = new String(data.getData(),"utf-8"))){
                switch (e.getType()){
                    case NODE_ADDED:
                        log.info("registry添加了节点【" + data.getPath() +"】, 添加的节点数据: ("  +dataStr );
                        localServerManager.addService(JSON.parseObject(dataStr , RegisterInfo.class) );
                        break;
                    case NODE_REMOVED:
                        log.info("registry移除了节点【" + data.getPath() +"】, 移除的节点数据: (" +dataStr);
                        localServerManager.removeService(JSON.parseObject(dataStr, RegisterInfo.class));
                        break;
                    case NODE_UPDATED:
           //             System.out.println("registry修改了节点【" + data.getPath() +"】, 修改的节点数据: (" +dataStr);
            //            break;
                    default:break;
                }
            }
        } );
        //添加客户端连接状态监听，用于 重连时 能重新想zookeeper注册节点，并及时拉去所有注册服务信息
        client.getConnectionStateListenable().addListener((client,state)->{
            switch (state){
                case RECONNECTED:
                    RegisterInfo registerInfo = SpringUtils.getBean(RegisterService.class).doRegister();
                    localServerManager.addService(registerInfo);
            }
        });
        try {
            treeCache.start();
            log.info("The listener is starting !");
            //   System.err.println("The listener is starting !");
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            System.err.println("监听器添加失败！");
            return false;
        }
    }

    private static String formatePath(String path){
        return path.indexOf('/')!=0?"/"+path:path;
    }






}
